package com.hwtx.soa.registry;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hzgj.bcl.soa.config.ReferenceDef;
import com.hzgj.bcl.soa.config.ServerDef;
import com.hzgj.bcl.soa.config.ServiceDef;
import com.hzgj.bcl.soa.registry.Consumer;
import com.hzgj.bcl.soa.registry.NotifyListener;
import com.hzgj.bcl.soa.registry.Provider;
import com.hzgj.bcl.soa.registry.Registry;
import com.hzgj.bcl.soa.registry.RegistryContext;
import com.hzgj.bcl.soa.registry.ServiceNotExistException;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.util.jmx.Management;
import com.hzgj.bcl.util.lang.StrKit;
import com.hzgj.bcl.util.lang.ThreadPoolKit;
import com.hzgj.bcl.util.reg.EtcdExtClient;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import lombok.Builder;
import lombok.Getter;
import mousio.etcd4j.promises.EtcdResponsePromise;
import mousio.etcd4j.responses.EtcdException;
import mousio.etcd4j.responses.EtcdKeysResponse;

import static com.hwtx.soa.registry.EtcdConstants.ATTR_AUTHOR;
import static com.hwtx.soa.registry.EtcdConstants.ATTR_GROUP;
import static com.hwtx.soa.registry.EtcdConstants.ATTR_TIME;
import static com.hwtx.soa.registry.EtcdConstants.getDownServicePath;
import static com.hwtx.soa.registry.EtcdConstants.getReferencePath;
import static com.hwtx.soa.registry.EtcdConstants.getServiceInfoPath;
import static com.hwtx.soa.registry.EtcdConstants.getServicePrefixPath;
import static com.hzgj.bcl.soa.registry.RegistryConstants.SERVICE_PREFIX;
import static com.hzgj.bcl.util.lang.ThreadPoolKit.getScheduledExecutorService;

public class EtcdRegistry implements Registry {

    protected static final Logger logger = LoggerFactory.getLogger(EtcdRegistry.class);
    protected int ttl;
    protected String mode;
    protected EtcdExtClient client;
    Map<String, ServiceDef> service_cache = Maps.newHashMap();
    ThreadPoolExecutor threadPoolExecutor = null;
    private Collection<ReferenceDef> refreshRefServices;

    public EtcdRegistry(EtcdExtClient client, int ttl, String mode) {
        this.client = client;
        this.ttl = ttl;
        this.mode = mode;
    }

    @Getter
    private ConcurrentMap<ReferenceDef, Set<NotifyListener>> subscribed = new ConcurrentHashMap<>();

    @Override
    public void subscribe(NotifyListener listener, ReferenceDef... referenceDefs) {
        for (ReferenceDef s : referenceDefs) {
            subscribed.compute(s, (k, v) -> {
                if (v == null) {
                    Set<NotifyListener> vv = Sets.newHashSet();
                    vv.add(listener);
                    return vv;
                } else {
                    v.add(listener);
                    return v;
                }
            });
        }


        if (refreshRefServices == null) {
            refreshRefServices = Sets.newHashSet();
        }
        refreshRefServices.addAll(Sets.newHashSet(referenceDefs));

        if (mode.equals(Constants.NOTIFY_CLIENT_PULL_MODE)) {
            ScheduledExecutorService scheduledExecutorService = getScheduledExecutorService("refresh-service-");
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                if (refreshRefServices != null && refreshRefServices.size() > 0) {

                    refreshRefServices.stream().filter(service -> {
                        if (StrKit.isBlank(service.getService())) {
                            logger.error("can't refresh service of " + service);
                            return false;
                        }
                        return true;

                    }).forEach(s -> notifyClient(s.getService()));
                }
            }, 0, ttl, TimeUnit.SECONDS);
        } else {
            new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        if (refreshRefServices != null && refreshRefServices.size() > 0) {
                            watchService();
                        }
                    } catch (Exception e) {
                        logger.error("{}", e);
                    }
                }
            }).start();
        }

        ScheduledExecutorService scheduledExecutorService = getScheduledExecutorService("registry-export-client-");
        Map<String, List<ReferenceDef>> clients = getClients(Arrays.asList(referenceDefs));
        if (clients.size() != 0) {
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                registryClients(clients);
            }, 0, ttl, TimeUnit.SECONDS);
        }
    }

    @Override
    public void unSubscribe(ReferenceDef... referenceDefs) {
        for (ReferenceDef s : referenceDefs) {
            subscribed.remove(s);
        }

        if (refreshRefServices != null) {
            refreshRefServices.removeAll(Sets.newHashSet(referenceDefs));
        }
    }

    @Override
    public boolean needLoopPullService() {
        return true;
    }

    @Override
    public void updateClient(Collection<ReferenceDef> referenceDefs) {
        registryClients(getClients(referenceDefs));
    }

    @Override
    public void notifyClient(String service) {
        String dir = getServicePrefixPath(service);
        try {
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDirWithRecursive(dir);
            if (nodes == null) {
                throw new ServiceNotExistException(service);
            }
            ServiceDef serviceDef = new ServiceDef();
            serviceDef.setName(service);
            List<ServerDef> servers = Lists.newArrayList();
            nodes.stream().forEach(etcdNode -> {
                List<Provider> providers = Lists.newArrayList();
                Object object = JSON.parse(etcdNode.value);
                if (object instanceof JSONObject) {
                    Provider provider = JSON.parseObject(etcdNode.value, Provider.class);
                    providers.add(provider);
                } else {
                    providers = JSON.parseArray(etcdNode.value, Provider.class);
                }
                int index = etcdNode.key.lastIndexOf("/");
                String ipAndPort = etcdNode.key.substring(index + 1);
                String[] address = ipAndPort.split(":");
                if (address != null && address.length == 2) {
                    int port = Integer.parseInt(address[1]);
                    providers.stream().forEach(provider -> {
                        ServiceDef.Attr newAttr = buildServiceDefAttr(provider, address[0], port);
                        newAttr.setInterface(service);
                        serviceDef.getAttrs().add(newAttr);
                        servers.add(newAttr.getServerDef());
                    });
                }
            });
            if (serviceDef.getAttrs().size() == 0) {
                service_cache.remove(service);
            } else {
                service_cache.put(service, serviceDef);
            }

            if (logger.isDebugEnabled()) {
                logger.debug("refresh service " + service);
            }

            if (subscribed.get(service) != null) {
                //没有初始化设置，避免服务端启动无用线程池
                if (servers.size() > 0) {
                    if (threadPoolExecutor == null) {
                        threadPoolExecutor = ThreadPoolKit.getThreadPool(4, 10, 3, TimeUnit.SECONDS, 30);
                    }
                    threadPoolExecutor.execute(() -> {
                        subscribed.get(service).stream().forEach(notifyListener -> notifyListener.notifyWithSingleService(servers));
                    });
                } else {
                    logger.warn("update server size is 0,client subscribe service is " + service);
                }
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
    }


    @Override
    public ServiceDef lookupService(String name) {
        ServiceDef serviceDef = service_cache.get(name);
        if (serviceDef == null) {
            notifyClient(name);
            serviceDef = service_cache.get(name);
        }
        return serviceDef;
    }

    @Override
    public void registryService(RegistryContext.ServerRegisterModel serverRegisterModel, boolean refreshTtl) {

        List<ServiceDef.Attr> services = serverRegisterModel.getServices().values().stream().filter(attr -> attr.isRegister()).collect(Collectors.toList());
        if (services.size() == 0) {
            if (logger.isDebugEnabled()) {
                logger.debug("server {} has no service to register.", serverRegisterModel.getServerId());
            }
            return;
        }
        registryService(services, refreshTtl);
    }

    @Deprecated
    private Consumer getConsumer(ReferenceDef referenceDef) {
        Consumer consumer = new Consumer();
        consumer.setGroup(referenceDef.getGroup());
        consumer.setVersion(referenceDef.getVersion());
        consumer.setService(referenceDef.getService());
        consumer.setHost(referenceDef.getHost());
        consumer.setJmxPort(Management.getJmxPort());
        return consumer;
    }

    private void watchService() {
        try {
            EtcdResponsePromise<EtcdKeysResponse> promise = client.waitForKey(SERVICE_PREFIX);
            EtcdKeysResponse response = promise.get();
            EtcdKeysResponse.EtcdNode etcdNode = response.getNode();
            String key = etcdNode.getKey();
            int _indexProvider = key.indexOf("/" + EtcdConstants.SERVICE_PROVIDERS);
            //之關注providers
            if (_indexProvider < 0) {
                return;
            }
            String service = key.substring(SERVICE_PREFIX.length() + 1, _indexProvider);
            if (subscribed.get(service) == null) {
                return;
            }
            ServiceDef.Attr _attr = null;
            NotifyListener.PushAction _action = NotifyListener.PushAction.set;
            switch (response.action) {
                case set:
                    _attr = getServiceAttr(service, key, etcdNode.getValue());
                    break;
                case expire:
                    _attr = getServiceAttr(service, key);
                    _action = NotifyListener.PushAction.expire;
                    break;
                case delete:
                    _attr = getServiceAttr(service, key);
                    _action = NotifyListener.PushAction.delete;
                    break;
            }

            if (_attr != null) {
                ServiceDef.Attr attr = _attr;
                NotifyListener.PushAction action = _action;
                subscribed.get(service).stream().forEach(notifyListener -> notifyListener.notifyClient(attr, action));
            }

        } catch (Exception e) {
            logger.error("{}", e);
        }
    }

    private ServiceDef.Attr getServiceAttr(String service, String key, String value) {

        Provider provider = JSON.parseObject(value, Provider.class);
        int index = key.lastIndexOf("/");
        String ipAndPort = key.substring(index + 1);
        String[] address = ipAndPort.split(":");
        if (address != null && address.length == 2) {
            int port = Integer.parseInt(address[1]);
            ServiceDef.Attr attr = buildServiceDefAttr(provider, address[0], port);
            attr.setInterface(service);
            return attr;
        } else {
            logger.error(" key {} is invalidate.", key);
        }
        return null;
    }

    private ServiceDef.Attr getServiceAttr(String service, String key) {
        int index = key.lastIndexOf("/");
        String ipAndPort = key.substring(index + 1);
        String[] address = ipAndPort.split(":");
        if (address != null && address.length == 2) {
            int port = Integer.parseInt(address[1]);
            ServiceDef.Attr attr = new ServiceDef.Attr();
            ServerDef serverDef = new ServerDef("");
            serverDef.setHost(address[0]);
            serverDef.setPort(port);
            attr.setServerDef(serverDef);
            serverDef.getServiceDefs().add(attr);
            attr.setInterface(service);
            return attr;
        } else {
            logger.error(" key {} is invalidate.", key);
        }
        return null;
    }

    private void registryService(List<ServiceDef.Attr> services, boolean refreshTtl) {

        List<String> keys = Lists.newArrayListWithCapacity(services.size());
        List<String> infos = Lists.newArrayListWithCapacity(services.size());
        if (refreshTtl) {
            for (ServiceDef.Attr attr : services) {
                keys.add(getServicePrefix(attr) + attr.getServerDef().getHost() + ":" + attr.getServerDef().getPort());
                infos.add(getServiceInfoPath(attr.getInterface()));
            }

            keys.forEach(key -> {
                try {
                    if (client.refresh(key, ttl)) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("refresh service provider " + key);
                        }
                    } else {
                        logger.info("refresh service provider failure," + key);
                    }
                } catch (EtcdException e1) {
                    logger.warn("refresh service failed, error: {}", e1);
                    if (e1.errorCode == 100) { //key not found
                        registryService(services, refreshTtl);
                    }
                } catch (Exception e) {
                    logger.error("{}", e);
                }
            });
            infos.forEach(key -> {
                try {
                    if (!client.refresh(key, ttl)) {
                        logger.error("refresh {}'serviceInfo failure", key);
                    }
                } catch (Exception e) {
                    logger.error("{}", e);
                }
            });
        } else {
            putService(services, refreshTtl);
        }
    }

    private void putService(List<ServiceDef.Attr> services, boolean refreshTtl) {
        Map<String, Provider> handleNodes = Maps.newHashMap();
        ConcurrentMap<String, ServiceDef.ServiceInfo> infos = Maps.newConcurrentMap();
        for (ServiceDef.Attr attr : services) {
            String key = getServicePrefix(attr) + attr.getServerDef().getHost() + ":" + attr.getServerDef().getPort();
            handleNodes.put(key, getProvider(attr));
            infos.putIfAbsent(getServiceInfoPath(attr.getInterface()), ServiceDef.getServiceInfo(attr.getInterface()));
        }
        for (Map.Entry<String, Provider> entry : handleNodes.entrySet()) {
            try {
                if (client.put(entry.getKey(), JSON.toJSONString(entry.getValue()), ttl)) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("registry service provider " + entry.getValue());
                    }
                } else {
                    logger.info("registry service provider failure," + entry.getValue());
                }
            } catch (EtcdException e1) {
                logger.warn("registry service failed, defs: {}, error: {}", entry.getValue(), e1);
                if (e1.errorCode == 100) { //key not found
                    registryService(services, refreshTtl);
                }
            } catch (Exception e) {
                logger.error("{}", e);
            }
        }
        infos.forEach((s, serviceInfo) -> {
            try {
                if (!client.put(s, JSON.toJSONString(serviceInfo), ttl)) {
                    logger.error("register serviceInfo {} failure", serviceInfo);
                }
            } catch (Exception e) {
                logger.error("{}", e);
            }
        });
    }

    private String getServicePrefix(ServiceDef.Attr attr) {
        if (client.exist(getDownServicePath(attr.getInterface()) + attr.getServerDef().getHost() + ":" + attr.getServerDef().getPort())) {
            if (logger.isDebugEnabled()) {
                logger.debug("service " + attr.getInterface() + " has been down.");
            }
            return getDownServicePath(attr.getInterface());
        } else {
            return getServicePrefixPath(attr.getInterface());
        }
    }

    private Provider getProvider(ServiceDef.Attr attr) {
        Map<String, String> settings = Maps.newHashMap();
        settings.put(ATTR_TIME, Long.toString(System.currentTimeMillis()));
        //@TODO:delete group
        settings.put(ATTR_GROUP, attr.getGroup());
        settings.put(ATTR_AUTHOR, attr.getAuthor());
        Provider provider = new Provider();
        provider.setAddress(attr.getServerDef().getHost() + ":" + attr.getServerDef().getPort());
        provider.setType("simple");
        provider.setName(attr.getInterface());
        provider.setVersion(attr.getVersion());
        provider.setGroup(attr.getGroup());
        provider.setNote(attr.getNote());
        provider.setSettings(settings);
        provider.setJmxPort(Management.getJmxPort());
        return provider;
    }

    private ConsumerNodeInfo getConsumerInfo(List<ReferenceDef> referenceDefs) {
        List<Consumer> consumers = Lists.newArrayList();
        String host = "";
        int jmxport = 0;
        for (ReferenceDef referenceDef : referenceDefs) {
            Consumer consumer = new Consumer();
            consumer.setGroup(referenceDef.getGroup());
            consumer.setVersion(referenceDef.getVersion());
            consumer.setService(referenceDef.getService());
            consumer.setHost(referenceDef.getHost());
            consumer.setJmxPort(Management.getJmxPort());
            consumers.add(consumer);
            jmxport = consumer.getJmxPort();
            host = consumer.getHost();
        }
        return ConsumerNodeInfo.builder().consumers(consumers).host(host).jmxPort(jmxport).build();
    }

    private ServiceDef.Attr buildServiceDefAttr(Provider provider, String ip, int port) {
        ServiceDef.Attr attrBuilder = new ServiceDef.Attr();
        if (provider.getSettings() != null) {
            attrBuilder.setGroup(provider.getSettings().get(ATTR_GROUP));
            if (provider.getSettings().get(ATTR_TIME) != null) {
                attrBuilder.setTime(new Date(Long.parseLong(provider.getSettings().get(ATTR_TIME))));
            }
        }
        attrBuilder.setNote(provider.getNote());
        attrBuilder.setVersion(provider.getVersion());
        ServerDef serverDef = new ServerDef(provider.getName());
        serverDef.setHost(ip);
        serverDef.setPort(port);
        attrBuilder.setServerDef(serverDef);
        attrBuilder.setVersion(provider.getVersion());
        String group = provider.getGroup();
        attrBuilder.setGroup(StrKit.isBlank(group) ? provider.getSettings().get(ATTR_GROUP) : group);
        serverDef.getServiceDefs().add(attrBuilder);
        return attrBuilder;
    }

    private void registryClients(Map<String, List<ReferenceDef>> referenceDefs) {
        referenceDefs.forEach((service, refs) -> {
            ConsumerNodeInfo info = getConsumerInfo(refs);
            String path = getReferencePath(service, info.host, info.jmxPort);
            try {
                if (client.put(path, JSON.toJSONString(info.consumers), ttl)) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("registry client consume " + info.consumers);
                    }
                } else {
                    logger.info("registry client consume failure," + info.consumers);
                }
            } catch (EtcdException e1) {
                logger.warn("registry client failed, defs: {}, error: {}", info.consumers, e1);
                if (e1.errorCode == 100) { //key not found
                    registryClients(referenceDefs);
                }
            } catch (Exception e) {
                logger.error("{}", e);
            }
        });
    }

    private Map<String, List<ReferenceDef>> getClients(Collection<ReferenceDef> referenceDefs) {
        Map<String, List<ReferenceDef>> clients = Maps.newHashMap();
        referenceDefs.stream().filter(referenceDef ->
                StrKit.isBlank(referenceDef.getUrl())
        ).forEach(referenceDef -> {
            if (referenceDef.isRegister()) {
                clients.compute(referenceDef.getInterfaceClass(), (key, refs) -> {
                    if (refs == null) {
                        refs = Lists.newArrayList();
                    }
                    refs.add(referenceDef);
                    return refs;
                });
            }
        });
        return clients;
    }


    @Builder
    static class ConsumerNodeInfo {
        public List<Consumer> consumers;
        public String host;
        public int jmxPort;
    }
}
